package com.atguigu.flink.streamapi.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * Created by Smexy on 2023/1/29
 *
 *      flink中输出:
 *              流.addSink(SinkFunction x)
 *
 *      向外部设备输出，有连接器(Connector)
 *
 *     -----------
 *      官方没有提供捆绑的连接器，社区爱好者提供了。
 *
 *      ------------
 *          String:  set key value
 *          Set：    sadd key member...
 *          List:  两端开口，可以从左侧或右侧添加
 *                  LPUSH key ele
 *                  RPUSH key ele
 *          Zset：  本质是set，每个member自带一个score，按照score排序
 *                  zadd  key score member score member...
 *          Hash：  value是Map结构
 *                      Map<field,value>
 *
 *                   hset key field1,value1
 *
 *
 */
public class Demo1_RedisSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                   /*
                    public RedisSink(
                      FlinkJedisConfigBase flinkJedisConfigBase,  配置Jedis连接池数据源的连接信息
                      RedisMapper<IN> redisSinkMapper   封装写操作命令
                      )
                 */

        FlinkJedisPoolConfig flinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
            .setMaxTotal(20)  //池子容量
            .setMinIdle(5)   //池子空闲时，依旧保留多少个激活的连接   5-10
            .setMaxIdle(10)
            .setHost("hadoop102")
            .setPort(6379)
            .setDatabase(0)
            .setTimeout(60000)
            .build();
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .addSink(new RedisSink(flinkJedisPoolConfig,new MyZsetWrite()));

                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    private static class MyZsetWrite implements RedisMapper<WaterSensor>{

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.ZADD,"wszset");
        }

        // element
        @Override
        public String getKeyFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }

        //分score
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.getVc().toString();
        }
    }

    private static class MyHashWrite implements RedisMapper<WaterSensor>{

        /*
                hset key field1,value1

                hset name key element

                we need hash name(key), hash key(field) and element(value).
                        additionalKey used as hash name for hash
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET,"wshash");
        }

        // 使用当前传感器的id作为field
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        // wszset = { field1 = v1,field2 =v2  }
        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }

    private static class MySetWrite implements RedisMapper<WaterSensor>{

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SADD);
        }

        //key: wsset
        @Override
        public String getKeyFromData(WaterSensor data) {
            return "wsset";
        }

        // [ v1,v2,v3 ]
        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }


    private static class MyListWrite implements RedisMapper<WaterSensor>{

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.LPUSH);
        }

        //key: watersensor
        @Override
        public String getKeyFromData(WaterSensor data) {
            return "watersensorlist";
        }

        // [ v1,v2,v3 ]
        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }

    //key: id  value: { }
    private static class MyStringWrite implements RedisMapper<WaterSensor>{

        /*
            返回RedisCommandDescription(封装写操作命令)

          hash或zset类型:
            public RedisCommandDescription(
                RedisCommand redisCommand,     写操作命令
                String additionalKey
                )

           set,string,list:   public RedisCommandDescription(RedisCommand redisCommand)


         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SET);
        }

        //从流中的一个数据中，获取key部分
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        //从流中的一个数据中，获取value部分
        @Override
        public String getValueFromData(WaterSensor data) {
            return JSON.toJSONString(data);
        }
    }
}
